package com.qingyunge.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.qingyunge.bean.SkuEvalInfoFinly;
import com.qingyunge.bean.TimeUserPingLunBean;
import com.qingyunge.util.DateFormatUtil;
import com.qingyunge.util.MyClickHouseUtil;
import com.qingyunge.util.MyKafkaUtil;
import com.qingyunge.util.MyfenPriceUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import java.time.Duration;

public class Dws_user_pingLun_time_window {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String topic = "sku_eval_topic";
        String gourpid = "Dws_user_pingLun_time";
        SingleOutputStreamOperator<TimeUserPingLunBean> timeUserPingLunBeanSingleOutputStreamOperator = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, gourpid))
                .flatMap(new FlatMapFunction<String, TimeUserPingLunBean>() {
                    @Override
                    public void flatMap(String s, Collector<TimeUserPingLunBean> collector) throws Exception {
                        try {
                            SkuEvalInfoFinly skuEvalInfoFinly = JSONObject.parseObject(s, SkuEvalInfoFinly.class);
                            String time = null;
                            String quarter = null;
                            try {
                                time = skuEvalInfoFinly.getDate().substring(0, 7);
                                quarter = time.substring(5, 7);
                            } catch (Exception e) {
                                time = DateFormatUtil.toYmdHms(System.currentTimeMillis()).substring(0, 7);
                                quarter = time.substring(5, 7);
                            }
                            Long pingLunNum = skuEvalInfoFinly.getEvalNum();
                            Double price = skuEvalInfoFinly.getPrice();
                            String salaryRange = MyfenPriceUtil.getSalaryRange(price);
                            String skuClass = skuEvalInfoFinly.getSkuId().substring(0, 1);
                            String location = skuEvalInfoFinly.getLocation();
                            Long ts = skuEvalInfoFinly.getTs();
                            collector.collect(new TimeUserPingLunBean(time, MyfenPriceUtil.getQuarter(Integer.parseInt(quarter)), salaryRange, location,
                                    skuClass, 1L, price, pingLunNum, ts));
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                });
        SingleOutputStreamOperator<TimeUserPingLunBean> timeUserPingWM = timeUserPingLunBeanSingleOutputStreamOperator.assignTimestampsAndWatermarks(WatermarkStrategy.<TimeUserPingLunBean>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<TimeUserPingLunBean>() {
            @Override
            public long extractTimestamp(TimeUserPingLunBean timeUserPingLunBean, long l) {
                return timeUserPingLunBean.getTs();
            }
        }));
        SingleOutputStreamOperator<TimeUserPingLunBean> reduceDS = timeUserPingWM.keyBy(new KeySelector<TimeUserPingLunBean, Tuple5<String, String, String, String, String>>() {
            @Override
            public Tuple5<String, String, String, String, String> getKey(TimeUserPingLunBean value) throws Exception {
                return Tuple5.of(value.getTime(), value.getQuarter(), value.getPriceQj(), value.getLocation(), value.getLocation());
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<TimeUserPingLunBean>() {
                    @Override
                    public TimeUserPingLunBean reduce(TimeUserPingLunBean timeUserPingLunBean, TimeUserPingLunBean t1) throws Exception {
                        timeUserPingLunBean.setNum(timeUserPingLunBean.getNum() + t1.getNum());
                        timeUserPingLunBean.setPingLunNum(timeUserPingLunBean.getPingLunNum() + t1.getPingLunNum());
                        timeUserPingLunBean.setPriceSum(timeUserPingLunBean.getPriceSum() + t1.getPriceSum());
                        return timeUserPingLunBean;
                    }
                });
        reduceDS.print("=====>");
        // TODO 写入clickhouse
        reduceDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_user_pingLun_time_table " +
                "values (?,?,?,?,?,?,?,?,?)"));
        // TODO 启动任务
        env.execute("Dws_user_pingLun_time_window");
    }
}
